package pers.wk.queues.direct;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import pers.wk.config.MQConfig;
import pers.wk.utils.MQConnectionUtils;

import java.io.IOException;

/**
 * 消息发布确认机制-异步confirm模式
 * <p>
 * Channel对象提供的ConfirmListener()回调方法只包含deliveryTag（当前Chanel发出的消息序号），
 * 我们需要自己为每一个Channel维护一个unconfirm的消息序号集合，
 * 每publish一条数据，集合中元素加1，每回调一次handleAck方法，
 * unconfirm集合删掉相应的一条（multiple=false）或多条（multiple=true）记录。
 * 从程序运行效率上看，这个unconfirm集合最好采用有序集合SortedSet存储结构
 */
@Slf4j
public class AsyncConfirmProducer {

    public static void main(String[] args) {
        //1.ConnectionFactory
        ConnectionFactory connectionFactory = MQConnectionUtils.getConnectionFactory("/");
        Connection connection = null;
        Channel channel = null;
        try {
            //2.创建连接
            connection = connectionFactory.newConnection();
            //3.通过连接获取通道Channel
            channel = connection.createChannel();
            //创建交换机
            channel.exchangeDeclare(MQConfig.ExchangeName.DIRECT_EXCHANGE, MQConfig.ExchangeType.direct);
            //创建队列
            channel.queueDeclare(MQConfig.QueueNames.EMAIL_QUEUE,
                    false, false, false, null);
            //队列绑定交换机
            String routingKey = "email";
            channel.queueBind(MQConfig.QueueNames.EMAIL_QUEUE, MQConfig.ExchangeName.DIRECT_EXCHANGE, routingKey);

            System.out.println("路由模式模式direct-生产者启动。。。");

            // 启用发布者确认模式 只能确定是否投递到了交换机，不能确认交换机投递到指定的队列中
            channel.confirmSelect();

            channel.addReturnListener(new ReturnListener() {
                @Override
                public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    log.info("send message error, replyCode: {}, replyText: {}, exchange: {}, routingKey: {}",
                            replyCode, replyText, exchange, routingKey);
                }
            });

            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    log.info("handleAck, deliveryTag: {}, multiple: {}", deliveryTag, multiple);
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    log.info("handleNack, deliveryTag: {}, multiple: {}", deliveryTag, multiple);
                }
            });

            for (int i = 0; i < 10; i++) {
                String message = "hello rabbitmq " + i;
                channel.basicPublish(MQConfig.ExchangeName.DIRECT_EXCHANGE, "routingKey", true, null, message.getBytes());
                log.info("send message, routingKey: {}, message: {}", routingKey, message);
            }

            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            MQConnectionUtils.close(connection, channel);
        }
    }
}
